<?php
/**
 * @Author: Marte
 * @Date:   2020-08-02 17:51:21
 * @Last Modified by:   Marte
 * @Last Modified time: 2020-08-02 08:19:18
 */

class Queue{
    protected $redis;
    protected $timeout = 5;
    protected $attempts = 3;

    // zrangebyscore delay -inf $time    获取到当前为止集合数据
    protected $lua = <<<lua
local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])
if next(val) ~= nil then
    redis.call('zremrangebyrank', KEYS[1], 0, #val-1)

    for i=1, #val, 100 do
        redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))
    end
end

return val
lua;

    static $queue = 'queue';
    static $delaySet = 'delay';
    static $reserveSet = 'reserve';

    public function __construct(){
        $this->redis = new \Redis();
        $this->redis->connect('127.0.0.1',6379);
    }

    /**
     * 添加普通任务
     * @param array  $data    
     */
    public function task(array $data)
    {
        $this->redis->rpush(self::$queue, json_encode($data));
    }
    
    /**
     * 添加延迟任务
     * @param  array  $data
     * @param  int  $time
     */
    public function delay(array $data, int $time)
    {
        $this->redis->zadd(self::$delaySet, $time, json_encode($data));
    }

    /**
     * 启动队列
     */
    public function listen(){
        while (true) {
        	//取出分数小于当前时间的delay集合，并加入到queue队列中
            $this->migrate([self::$delaySet, self::$queue, time()], 2);

            // $this->migrate([self::$reserveSet, self::$queue, time()], 2);
            //消费队列
            $this->consume();
        }
    }

    public function migrate($params, $keyCount){
    	return $this->redis->eval($this->lua, $params, $keyCount);
    }

    public function consume(){
    	//取出queue队列，重试次数自增并加到reserve集合
    	$lua = <<<lua
local job = redis.call('lpop', KEYS[1])
local reserved = false

if(job ~= false) then
	reserved = cjson.decode(job)
	reserved['attempts'] = reserved['attempts'] + 1
	reserved = cjson.encode(reserved)
	redis.call('zadd', KEYS[2], ARGV[1], reserved)
end

return {job, reserved}
lua;
		
        $nextJob = $this->redis->eval($lua, [self::$queue, self::$reserveSet, time() + $this->timeout], 2);
        //执行任务
        $this->deal($nextJob);
        // var_dump($nextJob);
    }

    public function deal($nextJob){
    	$taskJson = $nextJob[1] ?? '';
        if($taskJson){
        	$job = json_decode($taskJson, true);
        	//删除失败次数超过最大重试次数
        	if($job['attempts'] > $this->attempts){
        		$this->redis->zrem(self::$reserveSet, $taskJson);
        	}else{
        		$data = $job['data'];
        		//获取任务执行方法和参数
        		$func = $data['func'];
        		$params = $data['params'] ?? [];

        		$flag = true;
        		try{
        			$func(...$params);
        		}catch(\Exception $e){
        			//执行失败重回到delay队列
        			$flag = false;
        			$try = <<<lua
redis.call('zrem', KEYS[2], ARGV[1])
redis.call('zadd', KEYS[1], ARGV[2], ARGV[1])

return true
lua;
				
				var_dump('任务失败，重试次数' . $job['attempts']);
        		$this->redis->eval($try, [self::$delaySet, self::$reserveSet, $taskJson, time() + $this->timeout], 2);
        		}

        		//任务执行成功，删除reserve集合
        		if($flag){
        			$this->redis->zrem(self::$reserveSet, $taskJson);
        		}
        	}
        }
    }

    public function __destruct()
    {
        $this->redis->close();
    }
}

function able($str){
	var_dump($str);
}

function unable(){
	throw new Exception('error'); 
}

$lock = new Queue();
$lock->delay([
	'job' => 1, 
	'data' => [
		'func' => 'able',
		'params' => ['test']
	], 
	'attempts' => 0
], time() + 5);

$lock->task([
	'job' => 2, 
	'data' => [
		'func' => 'unable',
	], 
	'attempts' => 0
]);

$lock->listen();